//package com.sdgs.paas.sdk.core.mq;
//
//import java.io.UnsupportedEncodingException;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Map;
//import java.util.Map.Entry;
//import java.util.concurrent.ConcurrentHashMap;
//
//import javax.annotation.PreDestroy;
//
//import org.springframework.context.ApplicationContext;
//
//import com.ctg.mq.api.bean.ConsumeStatus;
//import com.ctg.mq.api.bean.MQMessage;
//import com.ctg.mq.api.bean.MQResult;
//import com.ctg.mq.api.bean.MQSendResult;
//import com.ctg.mq.api.exception.MQException;
//import com.ctg.mq.api.exception.MQProducerException;
//import com.ctg.mq.api.impl.MQConsumerImpl;
//import com.ctg.mq.api.impl.MQProducerImpl;
//import com.ctg.mq.api.impl.MQPushConsumerImpl;
//import com.sdgs.paas.sdk.core.log.PaasLogger;
//
///**
// * ctqmq操作对象,实现了{@link ICtgMqManager}
// *
// * @author wjj
// *
// */
//public class CtgMqManager implements ICtgMqManager {
//	
//	private ApplicationContext ctx;
//	private CtgMqConfig ctgMqConfig;
//
//	Map<String, MQProducerImpl> producerTopicMap = new ConcurrentHashMap<String, MQProducerImpl>();
//	List<MQProducerImpl> producerGroups = new ArrayList<MQProducerImpl>();
//	List<MQPushConsumerImpl> consumerGroups = new ArrayList<MQPushConsumerImpl>();
//	List<CtgConsumeListener> consumeListeners = new ArrayList<>();
//	Map<String, MQConsumerImpl> pullConsumerMap = new ConcurrentHashMap<String, MQConsumerImpl>();
//
//
//
//	// @PostConstruct
//	// public void init() throws MQException, ClassNotFoundException,
//	// InstantiationException, IllegalAccessException {
//	// Assert.notNull(ctgMqConfig, "配置文件未正确加载！");
//	// Assert.notNull(ctgMqConfig.getConsumerGroup(), "消费者配置未正确加载！");
//	// startConsumer(ctgMqConfig.getNamesrvAddr(),
//	// ctgMqConfig.getConsumerGroup(), ctgMqConfig.getAuthID(),
//	// ctgMqConfig.getAuthPWD(), ctgMqConfig.isConsumeOrderly(),
//	// ctgMqConfig.getBatchSize(),
//	// ctgMqConfig.getConsumerClientCallbackExecutorThreads(),
//	// ctgMqConfig.getConsumerClientWorkerThreads(),
//	// ctgMqConfig.getConsumerAbMsgProcNetCallbackExecutorThreads(),
//	// ctgMqConfig.getConsumerAbMsgProcNetWorkerThreads());
//	//// Assert.notNull(ctgMqConfig.getProducerGroup(), "生产者配置未正确加载！");
//	//// startProducer(ctgMqConfig.getNamesrvAddr(),
//	// ctgMqConfig.getProducerGroup(), ctgMqConfig.getAuthID(),
//	//// ctgMqConfig.getAuthPWD(), ctgMqConfig.getMaxMessageSize(),
//	// ctgMqConfig.getSendMsgTimeout(),
//	//// ctgMqConfig.getCompressMsgBodyOverHowmuch());
//	// }
//
//	public CtgMqManager(CtgMqConfig ctgMqConfig,ApplicationContext applicationContext) {
//		this.ctgMqConfig = ctgMqConfig;
//		this.ctx=applicationContext;
//	}
//
//	@Override
//	public boolean startConsumer(String namesrvAddr, List<ConsumerGroup> consumerGroup, String authID, String authPWD,
//			boolean consumeOrderly, int batchSize, int consumerClientCallbackExecutorThreads,
//			int consumerClientWorkerThreads, int consumerAbMsgProcNetCallbackExecutorThreads,
//			int consumerAbMsgProcNetWorkerThreads)
//			throws MQException, ClassNotFoundException, InstantiationException, IllegalAccessException {
//		boolean isConnected = false;
//
//		for (ConsumerGroup cg : consumerGroup) {
//			MQPushConsumerImpl consumer = new MQPushConsumerImpl(namesrvAddr, cg.getConsumerGroupName());
//
//			/**
//			 * 登录信息和消费模式设置
//			 */
//			// 连接时账号验证，目前ID和pwd不能为空，可以set，也可以配置在mq.properties里面
//			consumer.setAuthID(authID);
//			consumer.setAuthPWD(authPWD);
//			// 设置消费模式，包括（1）集群消费或广播消费（2）有序消费或无序消费。
//			// 默认是集群消费、无序消费； 队列消费必须设置成为有序消费模式
//			consumer.setConsumeOrderly(consumeOrderly); // 无序模式
//			// push消费一次拉取条数，默认为1
//			consumer.setBatchSize(batchSize);
//			consumer.setClientCallbackExecutorThreads(consumerClientCallbackExecutorThreads);
//			consumer.setClientWorkerThreads(consumerClientWorkerThreads);
//			consumer.setAbMsgProcNetCallbackExecutorThreads(consumerAbMsgProcNetCallbackExecutorThreads);
//			consumer.setAbMsgProcNetWorkerThreads(consumerAbMsgProcNetWorkerThreads);
//
//			/**
//			 * 建立连接, 启动消费者
//			 */
//			// try {
//			int result = consumer.connect();
//			if (result != 0) {
//				PaasLogger.warn("建立连接失败， 错误码是：{}", result);
//			}
//			// } catch (MQException e) {
//			// PaasLogger.error("建立连接异常",e);
//			// return isConnected;
//			// }
//
//			/**
//			 * 启动监听器
//			 */
//			for (final TopicDealer td : cg.getTopicDealer()) {
//				// 过滤条件
//				String subExpression = null;
//
//				CtgConsumeListener consumeListener=new CtgConsumeListener();
//				consumeListener.setStrategyClass(Class	.forName(td.getCtgMqConsumStrategy()));
//				consumeListener.setCtx(ctx);
//				consumeListeners.add(consumeListener);
//				
//				consumer.listenTopic(td.getTopic(), subExpression, consumeListener);
//
//			}
//			consumerGroups.add(consumer);
//
//		}
//
//		return isConnected;
//
//	}
//	
//	@Override
//	public void consumeStart(){
//		for ( CtgConsumeListener consumer : consumeListeners) {
//			consumer.startDealmsg();
//		}
//	}
//
//	@PreDestroy
//	@Override
//	public boolean stopConsumer() throws MQException {
//		for (MQPushConsumerImpl consumer : consumerGroups) {
//			consumer.close();
//		}
//
//		for (Entry<String, MQConsumerImpl> consumerEntry : pullConsumerMap.entrySet()) {
//			consumerEntry.getValue().close();
//		}
//
//		return true;
//	}
//
//	@Override
//	public boolean startProducer(String namesrvAddr, List<ProducerGroup> producerGroup, String authID, String authPWD,
//			int maxMessageSize, int sendMsgTimeout, int compressMsgBodyOverHowmuch) throws MQException {
//		boolean isConnected = false;
//
//		for (ProducerGroup pg : producerGroup) {
//			MQProducerImpl producer = new MQProducerImpl(namesrvAddr, pg.getProducerGroupName());
//
//			// 连接时账号验证，目前ID和pwd不能为空，可以set，也可以配置在mq.properties里面,暂不验证正确性！
//			producer.setAuthID(authID);
//			producer.setAuthPWD(authPWD);
//			// 客户端发送的最大消息大小,默认为128k
//			producer.setMaxMessageSize(maxMessageSize);
//			// 发送的超时时间时，默认为3s
//			producer.setSendMsgTimeout(sendMsgTimeout);
//			// 发送的压缩大小,可以不设置默认为4k
//			producer.setCompressMsgBodyOverHowmuch(compressMsgBodyOverHowmuch);
//
//			/**
//			 * 初始化
//			 */
//			int result = -1;
//			// try {
//			result = producer.connect();
//			// } catch (MQException e) {
//			// // Process connect exception
//			// e.printStackTrace();
//			// System.out.println("producer.connect exception, desc: " +
//			// e.getExpDesc());
//			// System.exit(1);
//			// }
//
//			if (result != 0) {
//				// Process connect error
//				MQException mqEx = new MQException(result, "producer.connect false!");
//				PaasLogger.error(mqEx, "producer.connect return: {}", result);
//				throw mqEx;
//			}
//
//			producerTopicMap.put(pg.getTopic(), producer);
//		}
//
//		return isConnected;
//	}
//
//	@PreDestroy
//	@Override
//	public boolean stopProducer() throws MQException {
//
//		/**
//		 * 关闭对象
//		 */
//		for (MQProducerImpl producer : producerTopicMap.values()) {
//			producer.close();
//		}
//
//		return true;
//
//	}
//
//	@Override
//	public boolean createMsg(String topic, String key, String tag, byte[] body) throws MQProducerException {
//		/**
//		 * 发送消息
//		 */
//		MQMessage message = new MQMessage(
//				// topic或者queue的名字
//				topic,
//				// key字段,一般为业务的关键词，可以将其设置成唯一的标识字段，如订单ID
//				key,
//				// 标签字段，方便过滤
//				tag,
//				// 消息体
//				body,
//				// 发往指定分区的关键词，可以为空，如果设置将发往其hash值的分区
//				getProducerFromTopic(topic).getProducerGroup());
//		// try {
//		MQSendResult sendResult = getProducerFromTopic(topic).send(message);
//		PaasLogger.info("消息发送成功，消息id：{}", sendResult.getMessageID());
//		// } catch (MQException e) {
//		// e.printStackTrace();
//		// System.out.println("producer.send exception, desc: " +
//		// e.getExpDesc());
//		// }
//		return true;
//	}
//
//	@Override
//	public void startPullConsumer(String namesrvAddr, String authID, String authPWD) throws MQException {
//
//		for (PullConsumerGroup pcg : ctgMqConfig.getPullConsumerGroup()) {
//			MQConsumerImpl consumer = new MQConsumerImpl(namesrvAddr, pcg.getGroupName());// 消费组
//			consumer.setAuthID(authID);
//			consumer.setAuthPWD(authPWD);
//			consumer.setClusterConsume(true);
//			int connectResult = consumer.connect();
//			if (connectResult != 0) {
//				throw new MQException(connectResult, String.format("消费组[%s]读取进程启动失败！", pcg.getGroupName()));
//			}
//			pullConsumerMap.put(pcg.getTopic(), consumer);
//		}
//
//	}
//
//	@Override
//	public List<String> pullMsg(String topic) throws MQException, InterruptedException, UnsupportedEncodingException {
//
//		MQConsumerImpl consumer = getPullConsumerFromTopic(topic);
//		List<String> msgs = new ArrayList<>();
//		List<MQResult> mqResultList = consumer.consumeMessagesByTopic(topic, // topic
//				null, // 订阅过滤表达式字符串
//				32, 3000// timeout
//		);
//		if (mqResultList.size() > 0) {
//			for (MQResult result : mqResultList) {
//				// paasPaasLogger.info("messageid:{}, messagekey:{}拉取成功！" +
//				// result.getMessage().getMessageID() ,
//				// result.getMessage().getKey());
//				consumer.ackMessage(result, ConsumeStatus.CONSUME_SUCCESS);
//				msgs.add(new String(result.getMessage().getBody(), "utf-8"));
//
//			}
//		}
//		return msgs;
//	}
//
//	private MQProducerImpl getProducerFromTopic(String topic) {
//		return producerTopicMap.get(topic);
//	}
//
//	private MQConsumerImpl getPullConsumerFromTopic(String topic) {
//		return pullConsumerMap.get(topic);
//	}
//
//	/**
//	 * @return the ctgMqConfig
//	 */
//	public CtgMqConfig getCtgMqConfig() {
//		return ctgMqConfig;
//	}
//
//	/**
//	 * @return the producerTopicMap
//	 */
//	public Map<String, MQProducerImpl> getProducerTopicMap() {
//		return producerTopicMap;
//	}
//
//	/**
//	 * @return the producerGroups
//	 */
//	public List<MQProducerImpl> getProducerGroups() {
//		return producerGroups;
//	}
//
//	/**
//	 * @return the consumerGroups
//	 */
//	public List<MQPushConsumerImpl> getConsumerGroups() {
//		return consumerGroups;
//	}
//
//	/**
//	 * @param ctgMqConfig
//	 *            the ctgMqConfig to set
//	 */
//	public void setCtgMqConfig(CtgMqConfig ctgMqConfig) {
//		this.ctgMqConfig = ctgMqConfig;
//	}
//
//	/**
//	 * @param producerTopicMap
//	 *            the producerTopicMap to set
//	 */
//	public void setProducerTopicMap(Map<String, MQProducerImpl> producerTopicMap) {
//		this.producerTopicMap = producerTopicMap;
//	}
//
//	/**
//	 * @param producerGroups
//	 *            the producerGroups to set
//	 */
//	public void setProducerGroups(List<MQProducerImpl> producerGroups) {
//		this.producerGroups = producerGroups;
//	}
//
//	/**
//	 * @param consumerGroups
//	 *            the consumerGroups to set
//	 */
//	public void setConsumerGroups(List<MQPushConsumerImpl> consumerGroups) {
//		this.consumerGroups = consumerGroups;
//	}
//
//	public ApplicationContext getCtx() {
//		return ctx;
//	}
//
//	public void setCtx(ApplicationContext ctx) {
//		this.ctx = ctx;
//	}
//
//	public List<CtgConsumeListener> getConsumeListeners() {
//		return consumeListeners;
//	}
//
//	public void setConsumeListeners(List<CtgConsumeListener> consumeListeners) {
//		this.consumeListeners = consumeListeners;
//	}
//
//	public Map<String, MQConsumerImpl> getPullConsumerMap() {
//		return pullConsumerMap;
//	}
//
//	public void setPullConsumerMap(Map<String, MQConsumerImpl> pullConsumerMap) {
//		this.pullConsumerMap = pullConsumerMap;
//	}
//
//
//}
